package net.bwie.jtp.dwd.order.job;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;

public class JtpOrderDetailDwdJob {

    public static void main(String[] args) throws Exception{
        // 创建tableEnv
        TableEnvironment tableEnv=getTableEnv();
        // 创建输入表-input
        createInputTable(tableEnv);
        // 处理数据-select
        Table resulTable = handle(tableEnv);
        // 创建输出表-output
        createOutputTable(tableEnv);
        //  保存数据
        saveToKafka(tableEnv,resulTable);
    }

    private static TableResult saveToKafka(TableEnvironment tableEnv, Table resulTable) {
        // 保存订单明细宽表数据到队列
        tableEnv.createTemporaryView("result_table", resulTable);
        tableEnv.executeSql("INSERT INTO dwd_order_detail_kafka_sink\n" +
                "SELECT\n" +
                "    id, order_id, user_id, order_status, sku_id, sku_name, province_id,\n" +
                "       activity_id, activity_rule_id, coupon_id, date_id, create_time,\n" +
                "       operate_date_id, operate_time, source_id, source_type, source_type_name,\n" +
                "       sku_num, split_original_amount, split_activity_amount, split_coupon_amount,\n" +
                "       split_total_amount, row_op_ts\n" +
                "FROM result_table"
        );
        TableResult tableResult = tableEnv.executeSql("SELECT*\n" +
                "FROM dwd_order_detail_kafka_sink");
        tableResult.print();
        return tableResult;
    }

    private static void createOutputTable(TableEnvironment tableEnv) {
        tableEnv.executeSql("CREATE TABLE dwd_order_detail_kafka_sink (\n" +
                "    id STRING,\n" +
                "    order_id STRING,\n" +
                "    user_id STRING,\n" +
                "    order_status STRING,\n" +
                "    sku_id STRING,\n" +
                "    sku_name STRING,\n" +
                "    province_id STRING,\n" +
                "    activity_id STRING,\n" +
                "    activity_rule_id STRING,\n" +
                "    coupon_id STRING,\n" +
                "    date_id STRING,\n" +
                "    create_time STRING,\n" +
                "    operate_date_id STRING,\n" +
                "    operate_time STRING,\n" +
                "    source_id STRING,\n" +
                "    source_type STRING,\n" +
                "    source_type_name STRING,\n" +
                "    sku_num STRING,\n" +
                "    split_original_amount STRING,\n" +
                "    split_activity_amount STRING,\n" +
                "    split_coupon_amount STRING,\n" +
                "    split_total_amount STRING,\n" +
                "    row_op_ts TIMESTAMP_LTZ(3),\n" +
                "    PRIMARY KEY (id) NOT ENFORCED\n" +
                ") WITH (\n" +
                "    'connector' = 'upsert-kafka',\n" +
                "    'topic' = 'dwd-order-detail',\n" +
                "    'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
                "    'key.format' = 'json',\n" +
                "    'value.format' = 'json'\n" +
                ")");

    }

    private static Table handle(TableEnvironment tableEnv) {
        //订单明细表order_detail
        Table orderDetailTable = tableEnv.sqlQuery(
                "SELECT\n" +
                        "    operate_date['id'] AS id,\n" +
                        "    operate_date['order_id'] AS order_id,\n" +
                        "    operate_date['sku_id'] AS sku_id,\n" +
                        "    operate_date['sku_name'] AS sku_name,\n" +
                        "    operate_date['create_time'] AS create_time,\n" +
                        "    operate_date['source_id'] AS source_id,\n" +
                        "    operate_date['source_type'] AS source_type,\n" +
                        "    operate_date['sku_num'] AS sku_num,\n" +
                        "    cast(\n" +
                        "      cast(operate_date['sku_num'] AS DECIMAL(16, 2)) * cast(operate_date['order_price'] AS DECIMAL(16, 2)) AS STRING\n" +
                        "    ) AS split_original_amount,\n" +
                        "    operate_date['split_total_amount'] AS split_total_amount,\n" +
                        "    operate_date['split_activity_amount'] AS split_activity_amount,\n" +
                        "    operate_date['split_coupon_amount'] AS split_coupon_amount,\n" +
                        "    proc_time\n" +
                        "FROM topic_db_kafka_source\n" +
                        "WHERE `table_name` = 'order_detail' AND operate_type = 'insert'"
        );
        tableEnv.createTemporaryView("order_detail_table", orderDetailTable);
        //订单表order_info
        Table orderInfoTable = tableEnv.sqlQuery(
                "SELECT\n" +
                "    operate_date['id'] AS id,\n" +
                "    operate_date['user_id'] AS user_id,\n" +
                "    operate_date['province_id'] AS province_id,\n" +
                "    operate_date['operate_time'] AS operate_time,\n" +
                "    operate_date['order_status'] AS order_status\n" +
                "FROM topic_db_kafka_source\n" +
                "WHERE `table_name` = 'order_info'\n" +
                "  AND (operate_type = 'insert' OR operate_type = 'update')"
        );
         tableEnv.createTemporaryView("order_info_table", orderInfoTable);
         //订单明细活动关联表数据：order_detail_activity
        Table orderDetailActivityTable = tableEnv.sqlQuery("SELECT\n" +
                "    operate_date['order_detail_id'] AS order_detail_id,\n" +
                "    operate_date['activity_id'] AS activity_id,\n" +
                "    operate_date['activity_rule_id'] AS activity_rule_id\n" +
                "FROM topic_db_kafka_source\n" +
                "WHERE `table_name` = 'order_detail_activity' AND operate_type = 'insert'\n"
        );
        tableEnv.createTemporaryView("order_detail_activity_table",orderDetailActivityTable);
        //订单明细优惠券关联表数据：order_detail_coupon
        Table orderDetailCouponTable = tableEnv.sqlQuery("SELECT\n" +
                "    operate_date['order_detail_id'] AS order_detail_id,\n" +
                "    operate_date['coupon_id'] AS coupon_id\n" +
                "FROM topic_db_kafka_source\n" +
                "WHERE `table_name` = 'order_detail_coupon' AND operate_type = 'insert'\n"
        );
        tableEnv.createTemporaryView("order_detail_coupon_table",orderDetailCouponTable);
        //字典维度表：dim_base_dic
        tableEnv.executeSql("CREATE TABLE dim_base_dic_hbase_source (\n" +
                "    row_key STRING,\n" +
                "    info ROW<dic_code STRING, dic_name STRING, parent_code STRING>,\n" +
                "    PRIMARY KEY (row_key) NOT ENFORCED\n" +
                ") WITH (\n" +
                "    'connector' = 'hbase-2.2',\n" +
                "    'table-name' = 'dim_base_dic',\n" +
                "    'zookeeper.quorum' = 'node101:2181,node102:2181,node103:2181',\n" +
                "    'lookup.async' = 'true',\n" +
                "    'lookup.cache.max-rows' = '10',\n" +
                "    'lookup.cache.ttl' = '1 hour'\n" +
                ")"
        );
//        tableEnv.executeSql("SELECT*FROM dim_base_dic_hbase_source").print();
//        5张表关联，获取订单明细事实宽表
        Table joinTable = tableEnv.sqlQuery("SELECT\n" +
                "    detail.id,\n" +
                "    detail.order_id,\n" +
                "    info.user_id,\n" +
                "    info.order_status,\n" +
                "    detail.sku_id,\n" +
                "    detail.sku_name,\n" +
                "    info.province_id,\n" +
                "    activity.activity_id,\n" +
                "    activity.activity_rule_id,\n" +
                "    coupon.coupon_id,\n" +
                "    date_format(detail.create_time, 'yyyy-MM-dd') AS date_id,\n" +
                "    detail.create_time,\n" +
                "    date_format(info.operate_time, 'yyyy-MM-dd') AS operate_date_id,\n" +
                "    info.operate_time,\n" +
                "    detail.source_id,\n" +
                "    detail.source_type,\n" +
                "    dic.dic_name AS source_type_name,\n" +
                "    detail.sku_num,\n" +
                "    detail.split_original_amount,\n" +
                "    detail.split_activity_amount,\n" +
                "    detail.split_coupon_amount,\n" +
                "    detail.split_total_amount,\n" +
                "    current_row_timestamp() AS row_op_ts\n" +
                "FROM order_detail_table detail\n" +
                "    JOIN order_info_table info ON detail.order_id = info.id\n" +
                "    LEFT JOIN order_detail_activity_table activity ON detail.id = activity.order_detail_id\n" +
                "    LEFT JOIN order_detail_coupon_table coupon ON detail.id = coupon.order_detail_id\n" +
                "    LEFT JOIN dim_base_dic_hbase_source FOR SYSTEM_TIME AS OF detail.proc_time AS dic\n" +
                "        ON detail.source_type = dic.row_key"
        );
        return joinTable;


    }

    private static void createInputTable(TableEnvironment tableEnv) {
        /*
        {
            "operate_type": "unknown",
                "operate_date": {
            "area_code": "320000",
                    "name": "江苏",
                    "region_id": "2",
                    "iso_3166_2": "CN-JS",
                    "id": 7,
                    "iso_code": "CN-32"
        },
            "db": "jtp_mall",
                "table": "base_province",
                "ts_ms": 1748851330975
        }
        */
        //DDL 语句，创建表
        tableEnv.executeSql("CREATE TABLE `topic_db_kafka_source`(\n" +
                    "`operate_type` STRING ,\n"+
                    "`operate_date` MAP<STRING,STRING>,\n"+
                    "`db_name` STRING,\n"+
                    "`table_name` STRING,\n"+
                    "`ts_ms` BIGINT,\n" +
                    "`proc_time` AS PROCTIME()\n" +
                    ") WITH (\n" +
                    "'connector' = 'kafka',\n" +
                    "'topic' = 'topic-db',\n" +
                    "'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n"+
                    "'properties.group.id'='gid-dwd-order-detail',\n"+
                    "'scan.startup.mode'='earliest-offset',\n"+
                    "'format'='json',\n"+
                    "'json.fail-on-missing-field' = 'false',\n" +
                    "'json.ignore-parse-errors'='true'\n"+
                    ")"
                );

    }

    private static TableEnvironment getTableEnv() {
        //  创建TableEnv
        EnvironmentSettings build = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();
        TableEnvironment tableEnv = TableEnvironment.create(build);
        //设置时区
        Configuration configuration = tableEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
        return tableEnv;

    }
}
